package com.beifeng.secondsort;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.protocol.datatransfer.Sender;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class MySecondSortMapReduce extends Configured implements  Tool {
	//KEYIN, VALUEIN, KEYOUT, VALUEOUT 传入的Key和Value类型，传出去的Key和Value类型
	static class MySecondSotMap extends Mapper<LongWritable, Text, SecondKey, NullWritable> {
		SecondKey map_Key = new SecondKey();
		NullWritable mr_value = NullWritable.get();

		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
		System.out.println("进入map");
		String line = 	value.toString();
		String[] words = line.split("\t");
		int year =  Integer.parseInt(words[0]);
		double temp =  Double.parseDouble(words[1]);
		String date = words[2];
		map_Key.setAll(year, temp, date);
		context.write(map_Key, mr_value);
		}
	}
	//KEYIN, VALUEIN, KEYOUT, VALUEOUT   输入keyvalue，输出keyvalue的类型
	static class MySecondSotReduce extends Reducer<SecondKey, NullWritable,SecondKey,NullWritable>{
		 private NullWritable result = NullWritable.get();
		@Override
		protected void reduce(SecondKey key, Iterable<NullWritable> values,
				Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
		//进入reduce
		System.out.println("进入reduce");
			context.write(key, result);
		}
	}
	
	public int run(String[] args) throws Exception {
		// TODO Auto-generated method stub
		//hadoop的八股文
		Configuration conf = this.getConf();
		Job job = Job.getInstance(conf, this.getClass().getSimpleName());
		//对job进行具体的配置
		
		//当你本地运行，这个设置可以不写，不会报错
		//当提价到集群上面运行的时候，这个设置不写，会报类找不到的异常 
		job.setJarByClass(MySecondSortMapReduce.class);
		
		//写一个输入路径
		Path input = new Path(args[0]);
		FileInputFormat.addInputPath(job, input);
		//写一个输出路径
		Path output = new Path(args[1]);
		FileOutputFormat.setOutputPath(job, output);
		
		//执行前先判断输出路径是否存在，存在就删除
		FileSystem fs = output.getFileSystem(conf);
		if(fs.exists(output)){
			fs.delete(output,true);
		}
		
		//设置运行的map类的相关参数
		job.setMapperClass(MySecondSotMap.class);
		job.setMapOutputKeyClass(SecondKey.class);
		job.setMapOutputValueClass(NullWritable.class);
		//==============shuffle=======================
		job.setPartitionerClass(MySecondSortPartition.class);
		job.setGroupingComparatorClass(MySecondSortGroup.class);
		
		//==============shuffle=======================
		//设置运行的Reduce的相关参数
		job.setReducerClass(MySecondSotReduce.class);
		job.setOutputKeyClass(SecondKey.class);
		job.setOutputValueClass(NullWritable.class);
		job.setNumReduceTasks(3);
		boolean isSuccess =  job.waitForCompletion(true);
		
		
		return isSuccess?0:1;
	}
	
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		Configuration conf = new Configuration();
		args = new String[]{
				"hdfs://centos6-1:8020/user/kewang/sort/datas/temp.txt",
				"hdfs://centos6-1:8020/user/kewang/sort/output"
		};
	/*	MyMapReduce test =  new MyMapReduce();
		try {
			test.run(args);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}*/
		try {
			ToolRunner.run(conf, new MySecondSortMapReduce(), args);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
 	}

}
